// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-05

package message

import (
	"context"
	"encoding/json"
	"gitee.com/go-libs/log"
	"gitee.com/gomq/sr/name"
	"net/http"
	"sync"
	"time"
)

var (
	messagePool = sync.Pool{
		New: func() interface{} {
			return &Message{}
		},
	}
	messageTime = time.Unix(0, 0)
)

// Message is a component that store received message attributes for
// servlet receiver to receive from given queues.
type Message struct {
	Body          []byte          `json:"Body,omitempty"`
	ConsumedCount int             `json:"ConsumedCount"`
	ConsumedTime  time.Time       `json:"ConsumedTime"`
	Ctx           context.Context `json:"-"`
	Delay         time.Duration   `json:"Delay"`
	Header        http.Header     `json:"Header,omitempty"`
	MessageId     string          `json:"MessageId,omitempty"`
	ProducedCount int             `json:"ProducedCount"`
	Queue         name.QueueName  `json:"Queue,omitempty"`
	Tag           name.TopicTag   `json:"Tag,omitempty"`
	Time          time.Time       `json:"Time,omitempty"`
	Topic         name.TopicName  `json:"Topic,omitempty"`
}

// NewMessage acquires an instance from pool, You must call Release()
// method to release it when done.
func NewMessage(ctx context.Context) *Message {
	return messagePool.
		Get().(*Message).
		acquirer().
		acquirerContext(ctx)
}

// NewMessageWithHeader acquires an instance from pool, You must call
// Release() method to release it when done.
func NewMessageWithHeader(header http.Header) *Message {
	return messagePool.
		Get().(*Message).
		acquirer().
		acquirerHeader(header)
}

// Release
// releases the message instance into pool.
func (o *Message) Release() {
	o.releaser()
}

// String
// converts the message as json string.
func (o *Message) String() string {
	if buf, err := json.Marshal(o); err == nil {
		return string(buf)
	}
	return "{}"
}

// + Access

func (o *Message) acquirer() *Message {
	o.ConsumedTime = time.Now()
	return o
}

func (o *Message) acquirerContext(ctx context.Context) *Message {
	return o.acquirerHeader(log.ContextHeader(ctx))
}

func (o *Message) acquirerHeader(header http.Header) *Message {
	if header != nil && len(header) > 0 {
		o.Header = header
	} else {
		o.Header = make(http.Header)
	}
	o.Ctx = log.NewContextWithHeader(o.Header)
	return o
}

func (o *Message) releaser() {
	o.Ctx, o.Body, o.Header, o.MessageId, o.Time, o.ConsumedTime = nil, nil, nil, "", messageTime, messageTime
	o.ConsumedCount, o.Delay, o.ProducedCount = 0, 0, 0
	o.Queue, o.Tag, o.Topic = "", "", ""
	messagePool.Put(o)
}
